package com.xian.scala.stream

import java.time.LocalDateTime
import java.time.format.DateTimeFormatter
import java.util.{Date, Locale}

import com.xian.scala.batch.Person
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.scala._

/**
 * scala版 flink 流处理
 */
object ScalaFlinkStreamTest {
  def main(args: Array[String]): Unit = {
    //获取流处理环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)





    val persons: DataStream[Person] = env.fromElements(
      Person(1, "Adam", new Date()),
      Person(3,"Sarah", new Date()),
      Person(2,"2Sarah", new Date())
    )


    persons.print()


    persons.map(p=>{
      p.id
      p.name
      p.birthday
      val localDateTime = LocalDateTime.parse(p.birthday.toString, DateTimeFormatter.ofPattern("EEE MMM dd HH:mm:ss z yyyy", Locale.US))
      val myDateTime = localDateTime.format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss SSS"))
      print(p.birthday+"||"+localDateTime+"||"+myDateTime+"\n")
      (p.id, p.name, myDateTime)

    }).print()
    /*格式化时间
    Person(1,Adam,Thu Jan 07 10:03:25 CST 2021)
    Thu Jan 07 10:03:25 CST 2021||2021-01-07T10:03:25||2021-01-07 10:03:25 000
    (1,Adam,2021-01-07 10:03:25 000)
     */




    env.execute("ScalaFlinkStreamTest")


  }

}
